package com.shujia.card;

import com.shujia.common.FlinkTool;
import com.shujia.tf.RealTImeCardDayFlowMapper;
import com.shujia.tf.RealTImeCardDayFlowToRedis;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class RealTImeCardDayFlow {
    public static void main(String[] args) throws Exception {
        //创建flink的环境
        StreamExecutionEnvironment env = FlinkTool.getFlinkEnv();

        //读取卡口过车数据

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092");
        properties.setProperty("group.id", "asdsadsa");

        //创建消费者

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
                "cars",
                new SimpleStringSchema(),
                properties);


        //读取数据
        DataStream<String> carsDS = env.addSource(flinkKafkaConsumer);


        /*
         * 1.1 实时统计每个卡扣当天总流量
         *
         */

        SingleOutputStreamOperator<Tuple3<Long, String, Long>> kvDS = carsDS.map(new RealTImeCardDayFlowMapper());


        ///安装卡口和时间分组
        KeyedStream<Tuple3<Long, String, Long>, Tuple2<Long, String>> keyedStream = kvDS.keyBy(new KeySelector<Tuple3<Long, String, Long>, Tuple2<Long, String>>() {
            @Override
            public Tuple2<Long, String> getKey(Tuple3<Long, String, Long> value) throws Exception {
                return Tuple2.of(value.f0, value.f1);
            }
        });


        //统计车流量
        SingleOutputStreamOperator<Tuple3<Long, String, Long>> carFlow = keyedStream.sum(2);


        //将数据保存到redis中
        carFlow.addSink(new RealTImeCardDayFlowToRedis());

        env.execute();

    }
}
